/*
 * Copyright 2019-2025 the original author or authors.
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * https://www.apache.org/licenses/LICENSE-2.0
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.iiifi.kite.rocketmq.core;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;

import com.iiifi.kite.rocketmq.exception.RocketMqException;

import lombok.extern.slf4j.Slf4j;

/**
 * RocketMQ的生产者的抽象基类
 *
 * @author kite@iiifi.com 花朝
 */
@Slf4j
public abstract class AbstractMqProducer {

    private static MessageQueueSelector messageQueueSelector = new SelectMessageQueueByHash();

    public AbstractMqProducer() {}

    @Autowired
    private DefaultMQProducer producer;

    /**
     * 同步发送消息
     *
     * @param message 消息体
     * @throws RocketMqException 消息异常
     */
    public void syncSend(Message message) throws RocketMqException {
        try {
            SendResult sendResult = producer.send(message);
            log.debug("send rocket-mq message ,messageId : {}", sendResult.getMsgId());
            this.doAfterSyncSend(message, sendResult);
        } catch (Exception e) {
            log.error("消息发送失败，topic : {}, msgObj {}", message.getTopic(), message);
            throw new RocketMqException("消息发送失败，topic :" + message.getTopic() + ",e:" + e.getMessage());
        }
    }


    /**
     * 同步发送消息
     *
     * @param message 消息体
     * @param hashKey 用于hash后选择queue的key
     * @throws RocketMqException 消息异常
     */
    public void syncSendOrderly(Message message, String hashKey) throws RocketMqException {
        if (StringUtils.isEmpty(hashKey)) {
            // fall back to normal
            syncSend(message);
        }
        try {
            SendResult sendResult = producer.send(message, messageQueueSelector, hashKey);
            log.debug("send rocket-mq message orderly ,messageId : {}", sendResult.getMsgId());
            this.doAfterSyncSend(message, sendResult);
        } catch (Exception e) {
            log.error("顺序消息发送失败，topic : {}, msgObj {}", message.getTopic(), message);
            throw new RocketMqException("顺序消息发送失败，topic :" + message.getTopic() + ",e:" + e.getMessage());
        }
    }

    /**
     * 重写此方法处理发送后的逻辑
     *
     * @param message 发送消息体
     * @param sendResult 发送结果
     */
    public void doAfterSyncSend(Message message, SendResult sendResult) {}

    /**
     * 异步发送消息
     *
     * @param message msgObj
     * @param sendCallback 回调
     * @throws RocketMqException 消息异常
     */
    public void asyncSend(Message message, SendCallback sendCallback) throws RocketMqException {
        try {
            producer.send(message, sendCallback);
            log.debug("send rocket-mq message async");
        } catch (Exception e) {
            log.error("消息发送失败，topic : {}, msgObj {}", message.getTopic(), message);
            throw new RocketMqException("消息发送失败，topic :" + message.getTopic() + ",e:" + e.getMessage());
        }
    }
}
